package kafkaStudy;

import java.util.Arrays;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

/** 
* @author 朱宏亮
* @version 创建时间：2017年12月17日 下午2:15:26 
* 类说明 kafka_2.11-1.0.0整合spring4.3.13 订阅者
*/
public class KafkaConsumer {

	@SuppressWarnings({ "unchecked", "resource", "rawtypes" })
	public static void main(String[] args){
			
			ApplicationContext factory = new ClassPathXmlApplicationContext("classpath:kafkaConsumer.xml");
			DefaultKafkaConsumerFactory ConFactory = (DefaultKafkaConsumerFactory) factory.getBean("consumerFactory");
			Consumer consumer = ConFactory.createConsumer();
			consumer.subscribe(Arrays.asList("zhlTopic"));
			while (true) {
	            ConsumerRecords<Integer, String> records = consumer.poll(100);
	               for (ConsumerRecord<Integer, String> record : records){
	            	   System.out.println("=============Consumer开始消费=============");
	                   String topic = record.topic();
	                   Integer key = record.key();
	                   String value = record.value();
	                   long offset = record.offset();
	                   int partition = record.partition();
	                   System.out.println("-------------topic:"+topic);
	                   System.out.println("-------------value:"+value);
	                   System.out.println("-------------key:"+key);
	                   System.out.println("-------------offset:"+offset);
	                   System.out.println("-------------partition:"+partition);
	                   System.out.println("~~~~~~~~~~~~~Consumer消费结束~~~~~~~~~~~~~");
	               }
	                  
	         }
		 }
}
